package day01;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.streaming.util.serialization.JSONKeyValueDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.io.IOException;
import java.util.Properties;

public class KafkaSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        Properties properties = new Properties();
        properties.setProperty("flink.partition-discovery.interval-millis", "30000");
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"s1.hadoop:9092");//集群的连接
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"group_33"); //消费组
       // env.addSource(new FlinkKafkaConsumer010<String>("topic_33",new SimpleStringSchema(),properties)).print();
        /**
         * 在读取kafka中数据的时候实现的反序列化接口存在两种
         *  DeserializationSchema<T> (只取到了kafka的value)  --》SimpleStringSchema 子 已经实现的反序列化工具
         *  KafkaDeserializationSchema<T>  (取到了更多的元数据和value) --》JSONKeyValueDeserializationSchema   JSON反序列化工具
         */
        //env.addSource(new FlinkKafkaConsumer010<ObjectNode>("topic_33", new JSONKeyValueDeserializationSchema(true),properties)).print();
        //env.addSource(new FlinkKafkaConsumer010<String>("topic_33",new MyDers01(),properties)).print();
       // env.addSource(new FlinkKafkaConsumer010<Apple>("topic_33",new MySer02(),properties)).print();
       // env.addSource(new FlinkKafkaConsumer010<String>("topic_33",new MySer03(),properties)).print();
        FlinkKafkaConsumer010<JSONObject> kafkaSource = new FlinkKafkaConsumer010<>("topic_33", new MyJsonDeser(true), properties);
       // kafkaSource.setStartFromEarliest();
       // kafkaSource.setStartFromGroupOffsets();
        kafkaSource.setStartFromLatest();
        env.addSource(kafkaSource).print();
        env.execute();
    }

    //String --> String
    public static class MyDers01 implements DeserializationSchema<String>{

        //String.getBytes --> Byte[]
        //new String(byte[]) --> String
        //二进制数组 》 输出
        @Override
        public String deserialize(byte[] bytes) throws IOException {
            return new String(bytes);
        }

        @Override
        public boolean isEndOfStream(String s) {
            return false;
        }

        //输出类型
        @Override
        public TypeInformation<String> getProducedType() {
            return TypeInformation.of(String.class);
        }
    }

   public static class MySer02 implements DeserializationSchema<Apple>{

       @Override
       public Apple deserialize(byte[] bytes) throws IOException {
           return new Apple(new String(bytes));
       }

       @Override
       public boolean isEndOfStream(Apple apple) {
           return false;
       }

       @Override
       public TypeInformation<Apple> getProducedType() {
           return TypeInformation.of(Apple.class);
       }
   }

   public static class MySer03 implements KafkaDeserializationSchema<String>{

       @Override
       public boolean isEndOfStream(String s) {
           return false;
       }

       @Override//()ConsumerRecord 对象
       public String deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
           String key=consumerRecord.key()==null?"":new String(consumerRecord.key());
           String value=new String(consumerRecord.value());
           long offset = consumerRecord.offset();
           String topic = consumerRecord.topic();
           int partition = consumerRecord.partition();
           return "key："+key+"value："+value+"offset："+offset+"topic："+topic+"partition："+partition;
       }

       @Override
       public TypeInformation<String> getProducedType() {
           return TypeInformation.of(String.class);
       }
   }

   //json形式的反序列化器将数据解析完毕放到一个json对象中 JSONObject -->fastJson
    public static class MyJsonDeser implements KafkaDeserializationSchema<JSONObject>{
        private Boolean hasMetaData=false;

       public MyJsonDeser(Boolean hasMetaData) {
           this.hasMetaData = hasMetaData;
       }

       @Override
       public boolean isEndOfStream(JSONObject jsonObject) {
           return false;
       }

       @Override
       public JSONObject deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
           //将key和value反序列化以后放入到一个jsonobject对象中，如果设置了元数据为true，就放入元数据
           JSONObject obj=new JSONObject();

           String key=consumerRecord.key()==null?"":new String(consumerRecord.key());
           String value=new String(consumerRecord.value());
           long offset = consumerRecord.offset();
           String topic = consumerRecord.topic();
           int partition = consumerRecord.partition();
           obj.put("key",key);
           obj.put("value",value);
           if(hasMetaData){
               obj.put("offset",offset);
               obj.put("topic",topic);
               obj.put("partition",partition);
           }
           return obj;
       }

       @Override
       public TypeInformation<JSONObject> getProducedType() {
           return TypeInformation.of(JSONObject.class);
       }
   }

}


class Apple{
    private String line;

    public String getLine() {
        return line;
    }

    public void setLine(String line) {
        this.line = line;
    }

    public Apple(String line) {
        this.line = line;
    }

    @Override
    public String toString() {
        return "Apple{" +
                "line='" + line + '\'' +
                '}';
    }
}
